package com.meiya.flink.util;

import com.meiya.flink.config.AppConfig;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ConsumerGroupListing;
import org.apache.kafka.clients.admin.ListConsumerGroupsResult;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * 连接测试工具类，用于测试各种连接
 */
public class ConnectionTestUtil {
    private static final Logger logger = LoggerFactory.getLogger(ConnectionTestUtil.class);

    /**
     * 测试Doris MySQL连接
     */
    public static void testDorisConnection() {
        logger.info("开始测试 Doris MySQL 连接...");
        String jdbcUrl = String.format("jdbc:mysql://%s/%s?user=%s&password=%s",
                AppConfig.getDorisMysqlNodes(), AppConfig.getDorisDatabase(),
                AppConfig.getDorisUsername(), AppConfig.getDorisPassword());

        try (Connection conn = DriverManager.getConnection(jdbcUrl)) {
            logger.info("Doris MySQL 连接成功！");

            // 执行简单查询
            try (Statement stmt = conn.createStatement()) {
                String sql = "SHOW DATABASES";
                try (ResultSet rs = stmt.executeQuery(sql)) {
                    logger.info("可用的数据库:");
                    while (rs.next()) {
                        logger.info(" - {}", rs.getString(1));
                    }
                }
            }
        } catch (Exception e) {
            logger.error("Doris MySQL 连接失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 测试Doris HTTP连接
     */
    public static void testDorisHttpConnection() {
        logger.info("开始测试 Doris HTTP API 连接...");
        HttpURLConnection conn = null;
        try {
            // 构建 HTTP API URL
            String httpUrl = "http://" + AppConfig.getDorisHttpNodes() + "/api/bootstrap";

            logger.info("尝试连接 Doris HTTP API: {}", httpUrl);

            URL url = new URL(httpUrl);
            conn = (HttpURLConnection) url.openConnection();
            conn.setRequestMethod("GET");
            conn.setConnectTimeout(5000);
            conn.setReadTimeout(5000);

            // 获取响应代码
            int responseCode = conn.getResponseCode();
            logger.info("Doris HTTP API 响应代码: {}", responseCode);

            if (responseCode == 200) {
                logger.info("Doris HTTP API 连接成功！");
                // 读取响应
                try (BufferedReader reader =
                             new BufferedReader(
                                     new InputStreamReader(conn.getInputStream()))) {
                    String line;
                    StringBuilder response = new StringBuilder();
                    while ((line = reader.readLine()) != null) {
                        response.append(line);
                    }
                    logger.info("Doris HTTP API 响应: {}", response.toString());
                }
            } else {
                logger.error("Doris HTTP API 连接失败，响应代码: {}", responseCode);
            }
        } catch (Exception e) {
            logger.error("Doris HTTP API 连接失败: {}", e.getMessage(), e);
        } finally {
            if (conn != null) {
                conn.disconnect();
            }
        }
    }

    /**
     * 测试Kafka连接
     */
    public static void testKafkaConnection() {
        logger.info("开始测试 Kafka 连接...");
        
        Properties props = AppConfig.getKafkaAdminProperties();
        
        try (AdminClient adminClient = AdminClient.create(props)) {
            logger.info("Kafka AdminClient 创建成功");
            
            // 列出所有主题
            ListTopicsResult topicsResult = adminClient.listTopics();
            logger.info("Kafka 主题列表: {}", topicsResult.names().get());
            
            // 列出所有消费者组
            ListConsumerGroupsResult groupsResult = adminClient.listConsumerGroups();
            logger.info("Kafka 消费者组列表:");
            for (ConsumerGroupListing group : groupsResult.all().get()) {
                logger.info(" - 消费者组: {}, 简单消费者: {}", group.groupId(), group.isSimpleConsumerGroup());
            }
            
            logger.info("Kafka 连接测试完成");
        } catch (InterruptedException | ExecutionException e) {
            logger.error("Kafka 连接测试失败: {}", e.getMessage(), e);
        }
    }
} 